package com.trade.module.database.operation;

import com.trade.common.utils.ResourceLoader;
import com.trade.module.database.basic.ConstConfig;
import com.trade.module.database.entity.QueryRecord;
import com.trade.module.database.meta.*;
import com.trade.module.database.service.QueryRecordService;
import com.trade.module.frontEndServer.FrontEndDBVo;
import lombok.extern.log4j.Log4j;
import net.sf.json.JSONArray;
import net.sf.json.JSONException;
import net.sf.json.JSONObject;

import oracle.sql.DATE;
import oracle.sql.TIMESTAMP;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang.StringUtils;

import java.lang.reflect.InvocationTargetException;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.CountDownLatch;

/**
 * Description:数据库数据查询逻辑处理类
 *
 * @JDK Version:1.8.0_40
 * Version: V1.0
 * Author: wangchun
 * Date: 2016/3/30 16:07
 */
@Log4j
public class DBDataQueryLoigic {

    private FrontEndDBVo frontEndDBVo;

    private ConstConfig.DBType dbType;
    private String dbIP;
    private String dbPort;
    private String dbName;
    private String userName;
    private String password;
    private List<String> tableNames;

    private DBQueryUtils dbQueryUtils;

    private int connPoolSize = 10;

    private DBDataQueryCallback dbDataQueryCallback;

    private String reqType = null;

    private Integer numPerPage = 100;

    private String authorizedUser;

    private String taskId;

    public DBDataQueryLoigic(FrontEndDBVo frontEndDBVo) {
        ResourceLoader resourceLoader = new ResourceLoader(ConstConfig.DDConfigFileName.DB_CONFIG);
        this.frontEndDBVo = frontEndDBVo;
        this.dbType = frontEndDBVo.getDbType();
        this.dbIP = frontEndDBVo.getDbUrl();
        this.dbPort = frontEndDBVo.getDbPort();
        this.dbName = frontEndDBVo.getDataBaseName();
        this.userName = frontEndDBVo.getDbUserName();
        this.password = frontEndDBVo.getDbPassword();
        this.tableNames = convertTableNameToList(frontEndDBVo.getDataTables());
        this.reqType = frontEndDBVo.getReqType();
        this.taskId = frontEndDBVo.getTaskId();
        this.authorizedUser = frontEndDBVo.getAuthorizedUser();
        this.connPoolSize = getPoolInitialSize(resourceLoader);
        dbQueryUtils = new DBQueryUtils(DBInfoUtil.getDBDriverName(dbType),
                DBInfoUtil.getDBUrl(dbType, dbIP, dbPort, dbName), userName, password, this.connPoolSize);
        this.numPerPage = getNumPerPage(resourceLoader);
    }

    /**
     * 获取配置中的连接池初始化size
     *
     * @param resourceLoader
     * @return
     */
    private Integer getPoolInitialSize(ResourceLoader resourceLoader) {
        Integer connPoolSize = 5;
        String poolInitalSizeStr = resourceLoader.getPropertiesValue(ConstConfig.DDConfigPropertiesKey.POOL_INITIAL_SIZE);
        if (poolInitalSizeStr != null) {
            connPoolSize = Integer.valueOf(poolInitalSizeStr);
        }
        return connPoolSize;
    }

    /**
     * 获取配置中的分页数
     *
     * @param resourceLoader
     */
    private Integer getNumPerPage(ResourceLoader resourceLoader) {
        Integer numPerPage = 100;
        String numPerPageStr = resourceLoader.getPropertiesValue(ConstConfig.DDConfigPropertiesKey.QUERY_NUM_PER_PAGE);
        if (numPerPageStr != null) {
            numPerPage = Integer.valueOf(numPerPageStr);
        }
        return numPerPage;
    }

    private List<String> convertTableNameToList(String tableNamesStr) {
        List<String> tableNamesList = null;
        if (org.apache.commons.lang.StringUtils.isNotEmpty(tableNamesStr)) {
            tableNamesList = Arrays.asList(tableNamesStr.split(ConstConfig.SeparatorChar.sTableSepar));
        }
        return tableNamesList;
    }

    /**
     * 开始查询任务
     *
     * @throws Exception
     */
    public void startQuery(String workId) throws Exception {
        //1.查看给定tableNames是否有指定要查那几张表的数据，如果没有指定要查的表，则需要查出数据库里所有的表名
        if (tableNames == null || tableNames.isEmpty()) {
            tableNames = new ArrayList<>();
            List<TableMetaData> tableMetaDataByDBName = dbQueryUtils.getTableMetaDataByDBName(dbName);
            for (TableMetaData tableMetaData : tableMetaDataByDBName) {
                tableNames.add(tableMetaData.getName());
            }
        }

        if (this.reqType == null) {
            return;
        }
        switch (this.reqType) {
            case ConstConfig.DBReqType.REQ_TABLE_CLOUMN:
                queryDBColumnMetaData();
                break;
            case ConstConfig.DBReqType.REQ_TABLE_DATA:
                queryDBData(workId);
                break;
            case ConstConfig.DBReqType.REQ_INSERT_DATA:
                insertDBData();
                break;
            default:
                break;
        }
    }

    private void insertDBData() throws SQLException {
        Map<String, List<JSONObject>> exportData = frontEndDBVo.getExportData();
        Set<Map.Entry<String, List<JSONObject>>> entries = exportData.entrySet();
        for (Map.Entry<String, List<JSONObject>> entry : entries) {
            String tableName = entry.getKey();
            List<JSONObject> value = entry.getValue();
            List<Map<String, String>> dataMapList = new ArrayList<>();
            for (JSONObject jsonObject : value) {
                Map<String, String> dataMap = jsonToHashMap(jsonObject);
                if (dataMap != null && dataMap.size() > 0) {
                    dataMapList.add(dataMap);
                }
            }
            dbQueryUtils.insertBatchData(tableName, dbName, dataMapList);
        }
        frontEndDBVo.setExportData(null);
        dbDataQueryCallback.dataPartial(frontEndDBVo);
        dbDataQueryCallback.dataFinished(frontEndDBVo);
        log.info("插入数据到目标库成功");
    }

    /**
     * @param jsonData
     * @func hashmap
     */
    public static Map<String, String> jsonToHashMap(JSONObject jsonData) {
        Map<String, String> rstList = new HashMap<>();
        try {
            for (Iterator<String> keyStr = jsonData.keys(); keyStr.hasNext(); ) {
                String columnName = keyStr.next().trim();
                if (StringUtils.equalsIgnoreCase("ROWKEY", columnName)) {
                    continue;
                }
                Object value = jsonData.get(columnName);
                if (value == null || value instanceof String) {
                    columnName = StringUtils.replaceChars(columnName, "BASE:", "");
                    rstList.put(columnName, (String) value);
                }
            }
        } catch (JSONException e) {
            e.printStackTrace();
        }
        return rstList;
    }

    /**
     * 查询表列名元数据
     */
    private void queryDBColumnMetaData() {
        Map<String, List<ColumnMetaData>> columnMetaDataMap = new HashMap<>();
        tableNames.forEach(tableName -> {
            try {
                List<ColumnMetaData> tableColumnMetaDatas = dbQueryUtils.getColumns(dbName, tableName);
                columnMetaDataMap.put(tableName, tableColumnMetaDatas);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        frontEndDBVo.setColumnMetaDataMap(columnMetaDataMap);
        dbDataQueryCallback.dataPartial(frontEndDBVo);
        dbDataQueryCallback.dataFinished(frontEndDBVo);
    }

    /**
     * 查询数据库数据
     */
    private void queryDBData(String workId) throws Exception {
        //2.对集合中的表启动查询
        Map<String, PageInfo> tableNamePagerDataMap = new HashMap<>();
        Map<String, String> tableNameSqlMap = new HashMap<>();

        //获取总页数
        int allPagesCount = 0;
        for (String tableName : tableNames) {
            //过滤已经查询过的数据
            String sql = getQuerySql(tableName, dbName, dbType, authorizedUser);
            PageInfo pageInfo = dbQueryUtils.queryPageinfo(tableName, 1, numPerPage, sql);
            tableNamePagerDataMap.put(tableName, pageInfo);
            tableNameSqlMap.put(tableName, sql);

            allPagesCount += pageInfo.getTotalPages();
        }

        //根据总页数协作完成工作
        CountDownLatch latch = new CountDownLatch(allPagesCount);

        for (String tableName : tableNames) {
            log.info("-------------------------table=" + tableName + "-----------------------------------");
            long currentPage = 1;
            //过滤已经查询过的数据
            String sql = tableNameSqlMap.get(tableName);
            log.info("querySql=" + sql);
            PageInfo pageInfo = tableNamePagerDataMap.get(tableName);
            long totalPages = pageInfo.getTotalPages();
            log.info("totalPages=" + totalPages);
            while (totalPages >= currentPage) {
                log.info("-------------------------table=" + tableName + "  currentPage=" + currentPage + "-----------------------------------");
                final long finalCurrentPage = currentPage;
                DBQueryThreadPool.getInstance().execute(workId,
                        new TaskRunnable(numPerPage, dbName, tableName, finalCurrentPage, sql, latch, pageInfo.getTotalRows())
              /*  () -> {
                            log.info("finalCurrentPage start=" + tableName + "     " + finalCurrentPage);
                            try {
                                queryTableData(numPerPage, dbName, tableName, finalCurrentPage, sql);
                            } catch (Exception e) {
                                dbDataQueryCallback.dataQueryException(e, frontEndDBVo);
                                log.error("queryDBData", e);
                            }
                            log.info("finalCurrentPage end=" + tableName + "     " + finalCurrentPage);
                            latch.countDown();
                        }*/
                );
                currentPage++;
            }
        }

        //没有新的类容需要查询的情况
        latch.await();

        //任务中所有线程结束后会调用后边的代码
        dbDataQueryCallback.dataFinished(frontEndDBVo);
        DBQueryThreadPool.getInstance().removeThreadPoolExecutor(workId);
    }

    /**
     * 拼装查询过滤sql
     *
     * @param tableName      表名
     * @param dbName
     * @param dbType
     * @param authorizedUser
     * @return
     */
    private String getQuerySql(String tableName, String dbName, ConstConfig.DBType dbType, String authorizedUser) throws SQLException, ParseException {
        String newTableName = tableName;
        if (dbType == ConstConfig.DBType.DAMENG) {
            newTableName = "\"" + dbName + "\".\"" + tableName + "\"";
        } else if (dbType == ConstConfig.DBType.ORACLE && StringUtils.isNotEmpty(authorizedUser)) {
            newTableName = "\"" + authorizedUser + "\".\"" + tableName + "\"";
        }
        StringBuffer sqlBuf = new StringBuffer("select * from " + newTableName + " where 1=1 ");

        Map<String, TableCloumnParam> tableCloumnParamMap = frontEndDBVo.getTableCloumnParamMap();
        if (tableCloumnParamMap != null && tableCloumnParamMap.containsKey(tableName)) {
            TableCloumnParam tableCloumnParam = tableCloumnParamMap.get(tableName);
            //（1）首先加入创建时间条件
            Date maxCreateTime = QueryRecordService.getInstance().getMaxCreateTime(this.dbName, tableName, taskId);
            Date maxUpdateTime = QueryRecordService.getInstance().getMaxUpdateTime(this.dbName, tableName, taskId);
            String dateFormat = "yyyy-MM-dd HH:mm:ss";
            String oracleDateFormat = "YYYY-MM-DD HH24:MI:SS";
            String createTimeCloumnName = tableCloumnParam.getCreateTimeCloumnName();
            String updateTimeCloumnName = tableCloumnParam.getUpdateTimeCloumnName();

            if (dbType == ConstConfig.DBType.DAMENG || dbType == ConstConfig.DBType.ORACLE) {
                createTimeCloumnName = "\"" + createTimeCloumnName + "\"";
                updateTimeCloumnName = "\"" + updateTimeCloumnName + "\"";
            }

            //获取表记录中当前数据中的增量时间戳最大值，锁定增量查询的范围，防止增量查询同时，数据不断再递增的情况数据出错
            if (StringUtils.isNotEmpty(tableCloumnParam.getCreateTimeCloumnName())) {
                buildMaxTimeCond(dbType, newTableName, sqlBuf, maxCreateTime, dateFormat, oracleDateFormat, createTimeCloumnName);
            }

            if (maxCreateTime != null && maxUpdateTime != null) {
                sqlBuf.append(" AND (" + createTimeCloumnName + " > ");
                buildTimeCond(dbType, sqlBuf, maxCreateTime, dateFormat, oracleDateFormat);
                sqlBuf.append(" OR " + updateTimeCloumnName + ">");
                buildTimeCond(dbType, sqlBuf, maxUpdateTime, dateFormat, oracleDateFormat);
                sqlBuf.append(" ) ");
                sqlBuf.append(" AND " + createTimeCloumnName + " is not null ");
                sqlBuf.append(" AND " + updateTimeCloumnName + " is not null ");
            } else {
                if (maxCreateTime != null) {
                    sqlBuf.append(" AND " + createTimeCloumnName + " > ");
                    buildTimeCond(dbType, sqlBuf, maxCreateTime, dateFormat, oracleDateFormat);
                    sqlBuf.append(" AND " + createTimeCloumnName + " is not null ");
                }
                //（2）首先加入修改时间条件
                if (maxUpdateTime != null) {
                    sqlBuf.append(" AND " + updateTimeCloumnName + ">");
                    buildTimeCond(dbType, sqlBuf, maxUpdateTime, dateFormat, oracleDateFormat);
                    sqlBuf.append(" AND " + updateTimeCloumnName + " is not null ");
                }
            }

            //（3）如果没有创建时间和修改时间则按ID来排除
            if (maxCreateTime != null || maxUpdateTime != null) {
                //添加排序条件
                addOrderByCreateTimeCond(sqlBuf, createTimeCloumnName, tableCloumnParam.getCreateTimeCloumnName(), dbType);
                return sqlBuf.toString();
            }

            List<String> pkCloumnNameList = tableCloumnParam.getPkCloumnNameList();
            if (pkCloumnNameList == null || pkCloumnNameList.size() <= 0) {
                //添加排序条件
                addOrderByCreateTimeCond(sqlBuf, createTimeCloumnName, tableCloumnParam.getCreateTimeCloumnName(), dbType);
                return sqlBuf.toString();
            }

            Map<String, List<String>> localIdRecord = getLocalIdRecord(tableName);
            if (localIdRecord.isEmpty()) {
                //添加排序条件
                addOrderByCreateTimeCond(sqlBuf, createTimeCloumnName, tableCloumnParam.getCreateTimeCloumnName(), dbType);
                return sqlBuf.toString();
            }

            Set<Map.Entry<String, List<String>>> localIdRecordEntrySet = localIdRecord.entrySet();
            localIdRecordEntrySet.forEach(localIdRecordEntry -> {
                String cloumnKey = localIdRecordEntry.getKey();
                List<String> cloumnValue = localIdRecordEntry.getValue();
                int size = cloumnValue.size();
                if (cloumnValue != null && size > 0) {
                    //oracle 对 in 中的长度有限制，最大为1000
                    if (dbType == ConstConfig.DBType.DAMENG) {
                        cloumnKey = "\"" + cloumnKey + "\"";
                    }
                    int limitSize = 1000;
                    int count = size / limitSize;
                    int countRemainder = size % limitSize;
                    for (int i = 0; i < count; i++) {
                        sqlBuf.append(" AND  ");
                        sqlBuf.append(cloumnKey);
                        sqlBuf.append(" NOT IN ");
                        sqlBuf.append(" (");
                        sqlBuf.append(StringUtils.join(cloumnValue.subList(i * limitSize, (i + 1) * limitSize), ","));
                        sqlBuf.append(") ");
                    }
                    if (countRemainder > 0) {
                        sqlBuf.append(" AND  ");
                        sqlBuf.append(cloumnKey);
                        sqlBuf.append(" NOT IN ");
                        sqlBuf.append(" (");
                        sqlBuf.append(StringUtils.join(cloumnValue.subList(size - countRemainder, size), ","));
                        sqlBuf.append(") ");
                    }
                }
            });

            //添加排序条件
            addOrderByCreateTimeCond(sqlBuf, createTimeCloumnName, tableCloumnParam.getCreateTimeCloumnName(), dbType);

        }

        return sqlBuf.toString();
    }

    private void buildMaxTimeCond(ConstConfig.DBType dbType, String newTableName, StringBuffer sqlBuf, Date maxLocalCreateTime, String dateFormat, String oracleDateFormat, String createTimeCloumnName) throws SQLException, ParseException {
        String maxRecordCreateTimeSql = "select max(" + createTimeCloumnName + ") from " + newTableName + " where 1=1 AND " + createTimeCloumnName + " is not null ";

        Object maxCreateTimeObj = dbQueryUtils.queryFieldMaxValue(maxRecordCreateTimeSql);
        Date maxCreateDate = null;
        if (maxCreateTimeObj != null) {
            if (maxCreateTimeObj instanceof TIMESTAMP) {
                TIMESTAMP ts = (TIMESTAMP) maxCreateTimeObj;
                maxCreateDate = ts.dateValue();
            } else if (maxCreateTimeObj instanceof DATE) {
                DATE ts = (DATE) maxCreateTimeObj;
                maxCreateDate = ts.dateValue();
            } else if (maxCreateTimeObj instanceof Timestamp) {
                Timestamp ts = (Timestamp) maxCreateTimeObj;
                maxCreateDate = ts;
            } else if (maxCreateTimeObj instanceof String) {
                maxCreateDate = DateUtil.stringToDate((String) maxCreateTimeObj, dateFormat);
            }
            if (maxCreateDate != null) {
                //为了防止源数据中有
                if (maxLocalCreateTime == null) {
                    sqlBuf.append(" AND ((" + createTimeCloumnName + " <= ");
                    buildTimeCond(dbType, sqlBuf, maxCreateDate, dateFormat, oracleDateFormat);
                    sqlBuf.append(" ) ");
                    sqlBuf.append(" OR ");
                    sqlBuf.append(createTimeCloumnName + " is null ");
                    sqlBuf.append(" ) ");
                } else {
                    if (maxLocalCreateTime.before(maxCreateDate)) {
                        sqlBuf.append(" AND (" + createTimeCloumnName + " <= ");
                        buildTimeCond(dbType, sqlBuf, maxCreateDate, dateFormat, oracleDateFormat);
                        sqlBuf.append(" ) ");
                    }
                }
            }
        }
    }

    /**
     * 添加按创建时间排序
     *
     * @param sqlBuf
     * @param createTimeCloumnName
     * @param createTimeCloumnNameTemp
     */
    private void addOrderByCreateTimeCond(StringBuffer sqlBuf, String createTimeCloumnName, String createTimeCloumnNameTemp, ConstConfig.DBType dbType) {
        //(mysql 在字段没有建立索引的情况下排序，查询效率急剧下降    /    sqlserver 中加入order by 会使查询的子查询会报错)
        //此字段未建立索引的情况下order by效率级低。
        if (StringUtils.isNotEmpty(createTimeCloumnNameTemp) && dbType != ConstConfig.DBType.MYSQL && dbType != ConstConfig.DBType.SQLSERVER
                && dbType != ConstConfig.DBType.MYSQL && dbType != ConstConfig.DBType.ORACLE) {
            sqlBuf.append(" order by " + createTimeCloumnName + " ");
        }
    }

    private void buildTimeCond(ConstConfig.DBType dbType, StringBuffer sqlBuf, Date dateTime, String dateFormat, String oracleDateFormat) {
        if (dbType == ConstConfig.DBType.ORACLE) {
            sqlBuf.append(" TO_TIMESTAMP(");
            sqlBuf.append("'");
            sqlBuf.append(DateUtil.dateToString(dateTime, dateFormat));
            sqlBuf.append("' ");
            sqlBuf.append(",");
            sqlBuf.append("'");
            sqlBuf.append(oracleDateFormat);
            sqlBuf.append("' ");
            sqlBuf.append(") ");
        } else {
            sqlBuf.append(" '");
            sqlBuf.append(DateUtil.dateToString(dateTime, dateFormat));
            sqlBuf.append("' ");
        }
    }

    /**
     * 将数据库中读取的主键记录，放入到map中
     *
     * @param tableName
     * @return
     * @throws SQLException
     */
    private Map<String, List<String>> getLocalIdRecord(String tableName) throws SQLException {
        Map<String, List<String>> pkIdMap = new HashMap();
        List<String> tableIdList = QueryRecordService.getInstance().getTableIdList(dbName, tableName, taskId);

        //表复合主键处理
        tableIdList.forEach(tableId -> {
            JSONArray jsonObjectArray = JSONArray.fromObject(tableId);
            for (int h = 0; h < jsonObjectArray.size(); h++) {
                JSONObject jsonObject = jsonObjectArray.getJSONObject(h);
                PKVo pkVo = (PKVo) jsonObject.toBean(jsonObject, PKVo.class);
                String cloumnName = pkVo.getCloumnName();
                String cloumnValue = pkVo.getCloumnValue();
                List<String> pkList = null;
                if (!pkIdMap.containsKey(cloumnName)) {
                    pkList = new ArrayList<>();
                    pkIdMap.put(cloumnName, pkList);
                } else {
                    pkList = pkIdMap.get(cloumnName);
                }
                pkList.add("'" + cloumnValue + "'");
            }
        });

        return pkIdMap;
    }

    public void queryTableData(int numPerPage, String dbName, String tableName, long currentPage, String sql) throws Exception {

        long startTime = System.currentTimeMillis();   //获取开始时间
        PageInfo pageInfoAndData = dbQueryUtils.queryTableDataByPaging(dbName, tableName, currentPage, numPerPage, sql);
        long endTime = System.currentTimeMillis();    //获取结束时间
        log.info("数据查询运行时长： " + (endTime - startTime) + "ms");

        Map<String, String> cloumnMap = new HashMap<>();

        FrontEndDBVo frontEndDBVoTemp = new FrontEndDBVo();
        BeanUtils.copyProperties(frontEndDBVoTemp, frontEndDBVo);

        //获取表字段参数(主键、创建时间，修改时间)
        Map<String, TableCloumnParam> tableCloumnParamMap = frontEndDBVoTemp.getTableCloumnParamMap();
        List<String> pkCloumnNameList = null;
        String createTimeCloumnName = null;
        String updateTimeCloumnName = null;
        TableCloumnParam tableCloumnParam = null;
        if (tableCloumnParamMap != null) {
            tableCloumnParam = tableCloumnParamMap.get(tableName);
            if (tableCloumnParam != null) {
                pkCloumnNameList = tableCloumnParam.getPkCloumnNameList();
                createTimeCloumnName = tableCloumnParam.getCreateTimeCloumnName();
                updateTimeCloumnName = tableCloumnParam.getUpdateTimeCloumnName();
            }
        }

        //循环多行数据
        List<QueryRecord> queryRecordList = new ArrayList<>();

        List<Map> resultList = pageInfoAndData.getResultList();

        if ((pkCloumnNameList != null && pkCloumnNameList.size() > 0) || StringUtils.isNotEmpty(createTimeCloumnName)
                || StringUtils.isNotEmpty(updateTimeCloumnName)) {
            long startTimeConvert = System.currentTimeMillis();   //获取开始时间
            for (Map tableData : resultList) {
                Set<Map.Entry> entrySet = tableData.entrySet();
                List<String> cloumnList = new ArrayList<String>();

                QueryRecord queryRecord = new QueryRecord();
                queryRecord.setTaskId(taskId);

                //防止有多个主键情况，所以采用分隔符存储
                List<PKVo> ids = new ArrayList();

                for (Map.Entry entry : entrySet) {
                    String keyCloumn = (String) entry.getKey();
                    if (!cloumnMap.containsKey(tableName)) {
                        cloumnList.add(keyCloumn);
                    }
                    Object value = entry.getValue();

                    //判断该字段是否为主键
                    for (String pkCloumnName : pkCloumnNameList) {
                        if (StringUtils.equalsIgnoreCase(pkCloumnName, keyCloumn)) {
                            ids.add(new PKVo(pkCloumnName, value.toString()));
                        }
                    }

                    if (createTimeCloumnName != null && StringUtils.equalsIgnoreCase(createTimeCloumnName, keyCloumn)) {
                        //判断是否为创建时间字段
                        Date date = convertObjectToDate(value);
                        queryRecord.setCreate_time(date);
                    } else if (updateTimeCloumnName != null && StringUtils.equalsIgnoreCase(updateTimeCloumnName, keyCloumn)) {
                        //判断是否为修改时间字段
                        Date date = convertObjectToDate(value);
                        queryRecord.setUpdate_time(date);
                    }
                }
                //设置表字段数据
                if (!cloumnMap.containsKey(tableName)) {
                    cloumnMap.put(tableName, StringUtils.join(cloumnList, ConstConfig.SeparatorChar.sColumSepar));
                }
                //设置queryRecord数据
                queryRecord.setTable_id(JSONArray.fromObject(ids).toString());
                queryRecord.setDb_name(this.dbName);
                queryRecord.setTable_name(tableName);
                queryRecord.setStatus("1");
                queryRecord.setQuery_time(new Date());

                queryRecordList.add(queryRecord);
            }
            long endTimeConvert = System.currentTimeMillis();    //获取结束时间
            log.info("数据转换运行时长： " + (endTimeConvert - startTimeConvert) + "ms");
        }

        //启动新线程保存数据到本地数据库
        if (queryRecordList != null && queryRecordList.size() > 0) {
            SaveRecordThreadPool.getInstance().execute(() -> {
                        try {
                            long startTimeSave = System.currentTimeMillis();   //获取开始时间
                            //存储查询过的记录到本地mysql数据里
                            QueryRecordService.getInstance().saveQueryRecordBatch(queryRecordList);

                            long endTimeSave = System.currentTimeMillis();    //获取结束时间
                            log.info("存储mongo数据运行时长： " + (endTimeSave - startTimeSave) + "ms");
                        } catch (Exception e) {
                            e.printStackTrace();
                            log.error("saveQueryRecordBatch:", e);
                        }
                    }
            );
        }

        log.info("put count=" + resultList.size());

        Map<String, List<Map>> resultMap = new HashMap<>();
        resultMap.put(tableName, resultList);
        //组装FrontEndDBVo

        frontEndDBVoTemp.setRecords(resultMap);
        frontEndDBVoTemp.setColumnMap(cloumnMap);
        frontEndDBVoTemp.setCurrentPage(currentPage);
        frontEndDBVoTemp.setTableName(tableName);
        dbDataQueryCallback.dataPartial(frontEndDBVoTemp);
    }

    public FrontEndDBVo queryTableDataSync(String tableName, long currentPage, String sql) throws Exception {

        long startTime = System.currentTimeMillis();   //获取开始时间
        PageInfo pageInfoAndData = dbQueryUtils.queryTableDataByPaging(dbName, tableName, currentPage, numPerPage, sql);
        long endTime = System.currentTimeMillis();    //获取结束时间
        log.info("数据查询运行时长： " + (endTime - startTime) + "ms");

        Map<String, String> cloumnMap = new HashMap<>();

        FrontEndDBVo frontEndDBVoTemp = new FrontEndDBVo();
        BeanUtils.copyProperties(frontEndDBVoTemp, frontEndDBVo);

        //获取表字段参数(主键、创建时间，修改时间)
        Map<String, TableCloumnParam> tableCloumnParamMap = frontEndDBVoTemp.getTableCloumnParamMap();
        List<String> pkCloumnNameList = null;
        String createTimeCloumnName = null;
        String updateTimeCloumnName = null;
        TableCloumnParam tableCloumnParam = null;
        if (tableCloumnParamMap != null) {
            tableCloumnParam = tableCloumnParamMap.get(tableName);
            if (tableCloumnParam != null) {
                pkCloumnNameList = tableCloumnParam.getPkCloumnNameList();
                createTimeCloumnName = tableCloumnParam.getCreateTimeCloumnName();
                updateTimeCloumnName = tableCloumnParam.getUpdateTimeCloumnName();
            }
        }

        //循环多行数据
        List<QueryRecord> queryRecordList = new ArrayList<>();

        List<Map> resultList = pageInfoAndData.getResultList();

        if ((pkCloumnNameList != null && pkCloumnNameList.size() > 0) || StringUtils.isNotEmpty(createTimeCloumnName)
                || StringUtils.isNotEmpty(updateTimeCloumnName)) {
            long startTimeConvert = System.currentTimeMillis();   //获取开始时间
            for (Map tableData : resultList) {
                Set<Map.Entry> entrySet = tableData.entrySet();
                List<String> cloumnList = new ArrayList<String>();

                QueryRecord queryRecord = new QueryRecord();
                queryRecord.setTaskId(this.taskId);

                //防止有多个主键情况，所以采用分隔符存储
                List<PKVo> ids = new ArrayList();

                for (Map.Entry entry : entrySet) {
                    String keyCloumn = (String) entry.getKey();
                    if (!cloumnMap.containsKey(tableName)) {
                        cloumnList.add(keyCloumn);
                    }
                    Object value = entry.getValue();

                    //判断该字段是否为主键
                    for (String pkCloumnName : pkCloumnNameList) {
                        if (StringUtils.equalsIgnoreCase(pkCloumnName, keyCloumn)) {
                            ids.add(new PKVo(pkCloumnName, value.toString()));
                        }
                    }

                    if (createTimeCloumnName != null && StringUtils.equalsIgnoreCase(createTimeCloumnName, keyCloumn)) {
                        //判断是否为创建时间字段
                        Date date = convertObjectToDate(value);
                        queryRecord.setCreate_time(date);
                    } else if (updateTimeCloumnName != null && StringUtils.equalsIgnoreCase(updateTimeCloumnName, keyCloumn)) {
                        //判断是否为修改时间字段
                        Date date = convertObjectToDate(value);
                        queryRecord.setUpdate_time(date);
                    }
                }
                //设置表字段数据
                if (!cloumnMap.containsKey(tableName)) {
                    cloumnMap.put(tableName, StringUtils.join(cloumnList, ConstConfig.SeparatorChar.sColumSepar));
                }
                //设置queryRecord数据
                queryRecord.setTable_id(JSONArray.fromObject(ids).toString());
                queryRecord.setDb_name(this.dbName);
                queryRecord.setTable_name(tableName);
                queryRecord.setStatus("1");
                queryRecord.setQuery_time(new Date());

                queryRecordList.add(queryRecord);
            }
            long endTimeConvert = System.currentTimeMillis();    //获取结束时间
            log.info("数据转换运行时长： " + (endTimeConvert - startTimeConvert) + "ms");
        }

        //启动新线程保存数据到本地数据库
        if (queryRecordList != null && queryRecordList.size() > 0) {
            SaveRecordThreadPool.getInstance().execute(() -> {
                        try {
                            long startTimeSave = System.currentTimeMillis();   //获取开始时间
                            //存储查询过的记录到本地mysql数据里
                            QueryRecordService.getInstance().saveQueryRecordBatch(queryRecordList);

                            long endTimeSave = System.currentTimeMillis();    //获取结束时间
                            log.info("存储mongo数据运行时长： " + (endTimeSave - startTimeSave) + "ms");
                        } catch (Exception e) {
                            e.printStackTrace();
                            log.error("saveQueryRecordBatch:", e);
                        }
                    }
            );
        }

        log.info("put count=" + resultList.size());

        Map<String, List<Map>> resultMap = new HashMap<>();
        resultMap.put(tableName, resultList);
        //组装FrontEndDBVo

        frontEndDBVoTemp.setRecords(resultMap);
        frontEndDBVoTemp.setColumnMap(cloumnMap);
        frontEndDBVoTemp.setCurrentPage(currentPage);
        frontEndDBVoTemp.setTableName(tableName);
        return frontEndDBVoTemp;
    }

    /**
     * 转化对象到java.util.Date
     *
     * @param value 需要转化的Object
     * @return
     */
    private Date convertObjectToDate(Object value) {
        Date date = null;
        if (value instanceof java.sql.Date) {
            date = (Date) value;
        } else if (value instanceof Timestamp) {
            date = new Date(((Timestamp) value).getTime());
        } else if (value instanceof TIMESTAMP) {
            try {
                date = ((TIMESTAMP) value).dateValue();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return date;
    }

    public DBQueryUtils getDBQueryUtils() {
        return dbQueryUtils;
    }

    public void setDbDataQueryCallback(DBDataQueryCallback dbDataQueryCallback) {
        this.dbDataQueryCallback = dbDataQueryCallback;
    }

    public void closeDataSource() throws SQLException {
        dbQueryUtils.closeDataSource();
    }


    class TaskRunnable implements Runnable {

        private int numPerPage;
        private String dbName;
        private String tableName;
        private long currentPage;
        private String sql;
        private CountDownLatch latch;
        private long totalRows;

        public TaskRunnable(int numPerPage, String dbName, String tableName, long currentPage, String sql, CountDownLatch latch, long totalRows) {
            this.numPerPage = numPerPage;
            this.dbName = dbName;
            this.tableName = tableName;
            this.currentPage = currentPage;
            this.sql = sql;
            this.latch = latch;
            this.totalRows = totalRows;
        }

        @Override
        public void run() {
            long startTime = System.currentTimeMillis();   //获取开始时间
            PageInfo pageInfoAndData = null;
            try {
                pageInfoAndData = dbQueryUtils.queryTableDataByPaging(dbName, tableName, currentPage, numPerPage, sql, totalRows);
            } catch (Exception e) {
                log.error("queryTableDataByPaging",e);
            }
            long endTime = System.currentTimeMillis();    //获取结束时间
            log.info("数据查询运行时长： " + (endTime - startTime) + "ms");

            Map<String, String> cloumnMap = new HashMap<>();

            FrontEndDBVo frontEndDBVoTemp = new FrontEndDBVo();
            try {
                BeanUtils.copyProperties(frontEndDBVoTemp, frontEndDBVo);
            } catch (IllegalAccessException e) {
                log.error("queryTableDataByPaging copyProperties",e);
            } catch (InvocationTargetException e) {
                log.error("queryTableDataByPaging copyProperties",e);
            }

            //获取表字段参数(主键、创建时间，修改时间)
            Map<String, TableCloumnParam> tableCloumnParamMap = frontEndDBVoTemp.getTableCloumnParamMap();
            List<String> pkCloumnNameList = null;
            String createTimeCloumnName = null;
            String updateTimeCloumnName = null;
            TableCloumnParam tableCloumnParam = null;
            if (tableCloumnParamMap != null) {
                tableCloumnParam = tableCloumnParamMap.get(tableName);
                if (tableCloumnParam != null) {
                    pkCloumnNameList = tableCloumnParam.getPkCloumnNameList();
                    createTimeCloumnName = tableCloumnParam.getCreateTimeCloumnName();
                    updateTimeCloumnName = tableCloumnParam.getUpdateTimeCloumnName();
                }
            }

            //循环多行数据
            List<QueryRecord> queryRecordList = new ArrayList<>();

            List<Map> resultList = pageInfoAndData.getResultList();

            if ((pkCloumnNameList != null && pkCloumnNameList.size() > 0) || StringUtils.isNotEmpty(createTimeCloumnName)
                    || StringUtils.isNotEmpty(updateTimeCloumnName)) {
                long startTimeConvert = System.currentTimeMillis();   //获取开始时间
                for (Map tableData : resultList) {
                    Set<Map.Entry> entrySet = tableData.entrySet();
                    List<String> cloumnList = new ArrayList<String>();

                    QueryRecord queryRecord = new QueryRecord();
                    queryRecord.setTaskId(taskId);

                    //防止有多个主键情况，所以采用分隔符存储
                    List<PKVo> ids = new ArrayList();

                    for (Map.Entry entry : entrySet) {
                        String keyCloumn = (String) entry.getKey();
                        if (!cloumnMap.containsKey(tableName)) {
                            cloumnList.add(keyCloumn);
                        }
                        Object value = entry.getValue();

                        //判断该字段是否为主键
                        for (String pkCloumnName : pkCloumnNameList) {
                            if (StringUtils.equalsIgnoreCase(pkCloumnName, keyCloumn)) {
                                ids.add(new PKVo(pkCloumnName, value.toString()));
                            }
                        }

                        if (createTimeCloumnName != null && StringUtils.equalsIgnoreCase(createTimeCloumnName, keyCloumn)) {
                            //判断是否为创建时间字段
                            Date date = convertObjectToDate(value);
                            queryRecord.setCreate_time(date);
                        } else if (updateTimeCloumnName != null && StringUtils.equalsIgnoreCase(updateTimeCloumnName, keyCloumn)) {
                            //判断是否为修改时间字段
                            Date date = convertObjectToDate(value);
                            queryRecord.setUpdate_time(date);
                        }
                    }
                    //设置表字段数据
                    if (!cloumnMap.containsKey(tableName)) {
                        cloumnMap.put(tableName, StringUtils.join(cloumnList, ConstConfig.SeparatorChar.sColumSepar));
                    }
                    //设置queryRecord数据
                    queryRecord.setTable_id(JSONArray.fromObject(ids).toString());
                    queryRecord.setDb_name(this.dbName);
                    queryRecord.setTable_name(tableName);
                    queryRecord.setStatus("1");
                    queryRecord.setQuery_time(new Date());

                    queryRecordList.add(queryRecord);
                }
                long endTimeConvert = System.currentTimeMillis();    //获取结束时间
                log.info("数据转换运行时长： " + (endTimeConvert - startTimeConvert) + "ms");
            }

            //启动新线程保存数据到本地数据库
            if (queryRecordList != null && queryRecordList.size() > 0) {
                SaveRecordThreadPool.getInstance().execute(() -> {
                            try {
                                long startTimeSave = System.currentTimeMillis();   //获取开始时间
                                //存储查询过的记录到本地mysql数据里
                                QueryRecordService.getInstance().saveQueryRecordBatch(queryRecordList);

                                long endTimeSave = System.currentTimeMillis();    //获取结束时间
                                log.info("存储mongo数据运行时长： " + (endTimeSave - startTimeSave) + "ms");
                            } catch (Exception e) {
                                e.printStackTrace();
                                log.error("saveQueryRecordBatch:", e);
                            }
                        }
                );
            }

            log.info("put count=" + resultList.size());

            Map<String, List<Map>> resultMap = new HashMap<>();
            resultMap.put(tableName, resultList);
            //组装FrontEndDBVo

            frontEndDBVoTemp.setRecords(resultMap);
            frontEndDBVoTemp.setColumnMap(cloumnMap);
            frontEndDBVoTemp.setCurrentPage(currentPage);
            frontEndDBVoTemp.setTableName(tableName);
            dbDataQueryCallback.dataPartial(frontEndDBVoTemp);

            //计数器减一
            latch.countDown();
        }
    }

}
